package com.opdar.athena.singletask.base;

import com.opdar.athena.base.RedisSessionManager;
import com.opdar.athena.client.agent.controllers.MigrateController;
import com.opdar.athena.client.agent.controllers.TransferController;
import com.opdar.athena.entities.TokenEntity;
import com.opdar.athena.messages.MigrateMessage;
import com.opdar.athena.messages.TransferMessage;
import com.opdar.athena.web.senders.ISender;
import org.springframework.beans.factory.annotation.Autowired;

import java.io.*;
import java.util.List;

/**
 * Created by shiju on 2017/8/16.
 */
public class SingleTaskSender implements ISender<Boolean> {
    private String defaultTopic = "TRANSFER";
    @Autowired
    private RedisSessionManager redisSessionManager;
    @Autowired
    private TransferController transferController;
    @Autowired
    private MigrateController migrateController;

    @Override
    public Boolean send(String key, byte[] value) {
        return this.send(defaultTopic,key,value);
    }

    @Override
    public Boolean  send(String topic, String key, byte[] value) {
        if(topic.equals("TRANSFER")){
            String messageId = new String(value);
            List<TokenEntity> tokens = redisSessionManager.getTokenWithUserId(key);
            for(TokenEntity entity:tokens){
                if(redisSessionManager.exist(entity.getToken())){
                    ByteArrayOutputStream baos = null;
                    DataOutputStream dataOutputStream = new DataOutputStream(baos = new ByteArrayOutputStream());
                    try {
                        byte[] messageData = messageId.getBytes();
                        byte[] tokenData = entity.getToken().getBytes();
                        dataOutputStream.writeInt(messageData.length);
                        dataOutputStream.write(messageData);
                        dataOutputStream.writeInt(tokenData.length);
                        dataOutputStream.write(tokenData);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    TransferMessage transferMessage = new TransferMessage(baos.toByteArray());
                    transferController.transfer(null,transferMessage);
                }else{
                    redisSessionManager.removeToken(entity.getToken(),entity.getUserId());
                }
            }
        }else if(topic.equals("MIGRATE")){
            DataInputStream dataInputStream = new DataInputStream(new ByteArrayInputStream(value));
            try {
                //将destination通知给要迁移的用户
                String id = dataInputStream.readUTF();
                String source = dataInputStream.readUTF();
                String destination = dataInputStream.readUTF();
                int type = dataInputStream.readInt();
                int status = dataInputStream.readInt();
                long num = dataInputStream.readLong();
                String token = dataInputStream.readUTF();

                ByteArrayOutputStream baos = null;
                DataOutputStream dataOutputStream = new DataOutputStream(baos = new ByteArrayOutputStream());
                try {
                    dataOutputStream.writeUTF(id);
                    dataOutputStream.writeUTF(destination);
                    dataOutputStream.writeInt(type);
                    dataOutputStream.writeLong(num);
                    dataOutputStream.writeUTF(token);
                    MigrateMessage migrateMessage = new MigrateMessage(baos.toByteArray());
                    migrateController.migrate(null,migrateMessage);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return true;
    }

}
